package com.leo.biz.controller;

import com.leo.base.Config;
import com.leo.biz.constants.ConfigCentre;
import com.leo.biz.po.CommMsg;
import com.leo.biz.service.cache.RedisUtil;
import com.leo.biz.service.impl.Alog;
import com.leo.biz.service.kafkaclient.KafkaService;
import com.leo.biz.service.msgcenter.MsgRecvService;
import com.leo.biz.service.msgcenter.MsgSendService;
import com.leo.biz.service.pub.SCaptcha;
import com.leo.biz.service.pub.TypeUtils;
import com.leo.biz.service.zookeeper.ZkClient;
import kafka.message.MessageAndMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.servlet.http.Cookie;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.http.HttpSession;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by leo on 2016/10/20.
 */
@Controller
@RequestMapping(value = "/test")
public class TestController {
    @Autowired
    private Alog alog;

    @Autowired
    private MsgRecvService msgRecvService;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private MsgSendService msgSendService;

    static Logger logger = LoggerFactory.getLogger(TestController.class);

    private static AtomicInteger atomicCount=new AtomicInteger(0);

    @Config(name = "redis.port")
    private static String testconfig;



    @RequestMapping(value="/dotest", method= RequestMethod.GET,produces = "application/json; charset=utf-8")
    public void test(HttpSession session,HttpServletResponse response)
    {

        System.out.println(testconfig);
        RedisUtil.putKey("ABC",testconfig);

        logger.info(RedisUtil.getKey("ABC").toString());



        response.setContentType("image/jpeg");
        // 禁止图像缓存。
        response.setHeader("Pragma", "no-cache");
        response.setHeader("Cache-Control", "no-cache");
        response.setDateHeader("Expires", 0);
        SCaptcha instance = new SCaptcha();
        Cookie cookie = new Cookie("scaptcha", instance.getCode());
        cookie.setMaxAge(1800);
        response.addCookie(cookie);
        try {
            instance.write(response.getOutputStream());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @RequestMapping(value="/clear", method= RequestMethod.GET,produces = "application/json; charset=utf-8")
    @ResponseBody
    public Object clear(HttpSession session)
    {
        //InputStream ipt=TestController.class.getResourceAsStream("/log4j.properties");
        //PropertyConfigurator.configure(ipt);
        atomicCount.set(0);
        try {
            Thread.sleep(100L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info(" count: {}", atomicCount.get());
        return "count:0";
    }

    @RequestMapping(value="/getrediskey/{key}", method= RequestMethod.GET,produces = "application/json; charset=utf-8")
    @ResponseBody
    public Object getkey(HttpSession session,@PathVariable String key)
    {
        //InputStream ipt=TestController.class.getResourceAsStream("/log4j.properties");
        //PropertyConfigurator.configure(ipt);
        return  RedisUtil.getKey(key);
    }

    @RequestMapping(value="/delrediskey/{key}", method= RequestMethod.GET,produces = "application/json; charset=utf-8")
    @ResponseBody
    public Object delrediskey(HttpSession session,@PathVariable String key)
    {
        //InputStream ipt=TestController.class.getResourceAsStream("/log4j.properties");
        //PropertyConfigurator.configure(ipt);
        return  RedisUtil.del(key);
    }

    @RequestMapping(value="/queryzkcfg/{key}", method= RequestMethod.GET,produces = "application/json; charset=utf-8")
    @ResponseBody
    public Object querycfg(HttpSession session,@PathVariable String key)
    {
        if(key.equalsIgnoreCase("0"))
        {
            try {
                return ZkClient.getActiveServiceNode(ConfigCentre.getServiceCfgPath());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        //InputStream ipt=TestController.class.getResourceAsStream("/log4j.properties");
        //PropertyConfigurator.configure(ipt);
        return  "OK";
    }

    @RequestMapping(value="/produce/{topicid}/{key}/{msg}", method= RequestMethod.GET,produces = "application/json; charset=utf-8")
    @ResponseBody
    public Object produce(HttpSession session,@PathVariable String topicid,@PathVariable String key,@PathVariable String msg)
    {
        CommMsg commMsg=new CommMsg();
        commMsg.setSender("MaGicVV");
        commMsg.setRecver("Honor");
        commMsg.setContent(msg);
        //System.out.println(TypeUtils.object2Json(commMsg));
        msgSendService.send(topicid, key,TypeUtils.object2Json(commMsg));
        return  "OK";
    }

    @RequestMapping(value="/consume/{topicid}", method= RequestMethod.GET,produces = "application/json; charset=utf-8")
    @ResponseBody
    public Object consume(HttpSession session,@PathVariable String topicid)
    {
        msgRecvService.start(topicid,new MsgRecvService.IRecvDataCallBack() {
            @Override
            public void handle(MessageAndMetadata data) {
                System.out.println(Thread.currentThread().getName()+" | "+new String((byte[])data.message())+" | "+data.topic()+" | "+data.partition()+" | "+data.offset());
                //CommMsg commMsg=(CommMsg)TypeUtils.json2Object(new String((byte[])data.message()),CommMsg.class);
               // System.out.println(commMsg.getSender()+" "+commMsg.getRecver());
            }
        });
        return  "OK";
    }

}
